package com.chukun.flink.stream.sideoutput;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author chukun
 * @version 1.0.0
 * @description 侧端输出基本操作
 * @createTime 2022年05月27日 00:44:00
 */
public class SideOutputOperator {

    public static final String[] DATA = new String[]{
            "In addition to the main stream that results from DataStream operations",
            "When using side outputs",
            "We recommend you",
            "you first need to define an OutputTag that will be used to identify a side output stream"
    };


    final static OutputTag<String> REJECTED_WORDS_TAG = new OutputTag<String>("rejected") {
    };

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> inputStream = env.fromElements(DATA);

        SingleOutputStreamOperator<String> processStream = inputStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String in, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                if (in.length() < 20) {
                    //不符合条件的数据侧端输出
                    context.output(REJECTED_WORDS_TAG, in);
                } else {
                    // 正常数据
                    collector.collect(in);
                }
            }
        });

        DataStream<String> sideOutput = processStream.getSideOutput(REJECTED_WORDS_TAG);

        sideOutput.print("侧端输出");

        // 正常数据处理
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = processStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return Tuple2.of(s, s.length());
            }
        });

        mapStream.print("正常数据处理");

        env.execute("SideOutputOperator");
    }
}
